#include "tcp_connection.h"
#include <iostream>

namespace rr {
namespace nio {

static const uint32_t kBufferDefaultSize = 64*1024;
static const uint32_t kBufferMaxSize = 2*1024*1024;

TcpConnection::TcpConnection(EventLoop::Var loop, std::shared_ptr<uv_tcp_t> tcp_handle)
: loop_(loop)
, tcp_handle_(tcp_handle)
{
    tcp_handle_->data = this;
    printf("TcpConnection::TcpConnection() this[%p]\n", this);
}

TcpConnection::~TcpConnection()
{
    printf("TcpConnection::~TcpConnection() this[%p]\n", this);
}


int32_t TcpConnection::StartRead()
{
    read_buffer_ = std::make_shared<Buffer>(kBufferDefaultSize, kBufferMaxSize);
    int32_t ret = uv_read_start((uv_stream_t *)tcp_handle_.get(), OnAllocBuffer, OnRead);
    if(ret != 0) {
        printf("TcpConnection::StartRead ret=%d error: %s\n", ret, uv_strerror(ret));
        return -1;
    }

    return 0;
}

int32_t TcpConnection::Close()
{
    printf("TcpConnection::Close this:%p\n", this);
    SetConState(kDisconnecting);
    uv_handle_t *handle = (uv_handle_t *)tcp_handle_.get();
    if(!uv_is_closing((uv_handle_t *)handle)) {
        uv_close((uv_handle_t*)handle, TcpConnection::OnCloseComplete);
    } else {
        DoCloseComplete();
    }

    return 0;
}

int32_t TcpConnection::Send(const Buffer::Sptr& buffer, const WriteCallback& cb, void *user)
{
    WriteReqSptr req = std::make_shared<WriteReq>();
    req->uv_req.data = this;
    req->uv_buf.base = (char *)buffer->Peek();
    req->uv_buf.len = buffer->ReadableBytes();
    req->cb = cb;
    req->user_data = user;
    req->buffer = buffer;

    return Write(req);
}

int32_t TcpConnection::SendNoCopy(const char *data, uint32_t size, const WriteCallback& cb, void *user)
{
    WriteReqSptr req = std::make_shared<WriteReq>();
    req->uv_req.data = this;
    req->uv_buf.base = (char *)data;
    req->uv_buf.len = size;
    req->cb = cb;
    req->user_data = user;
    req->buffer = nullptr;

    return Write(req);
}

void TcpConnection::OnAllocBuffer(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    printf("OnAllocBuffer suggested_size=%zu\n", suggested_size);

    TcpConnection *ptr = (TcpConnection *)handle->data;
    size_t alloc_len = suggested_size;

    do
    {
        if(ptr->read_buffer_->EnsureWritableBytes(suggested_size)) {
            break;
        }
        printf("OnAllocBuffer faild suggested_size=%zu\n", suggested_size);
        ptr->read_buffer_->Clear();
        ptr->read_buffer_->EnsureWritableBytes(suggested_size);
        size_t write_able = ptr->read_buffer_->WritableBytes();
        alloc_len = suggested_size <= write_able ? suggested_size : write_able;
    } while (0);
    
    buf->base = (char *)ptr->read_buffer_->BeginWrite();
    buf->len = alloc_len;

    return;
}

void TcpConnection::OnRead(uv_stream_t *handle, ssize_t nread, const uv_buf_t *buf)
{
    TcpConnection *ptr = (TcpConnection *)handle->data;
    ptr->HandleRead(nread);

    return;
}

void TcpConnection::OnCloseComplete(uv_handle_t *handle)
{
    printf("TcpConnection::OnCloseComplete handle:%p data:%p\n", handle, handle->data);
    TcpConnection *ptr = (TcpConnection *)handle->data;
    ptr->loop_->GetEventRecorder()->RemoveRecord(handle);
    ptr->DoCloseComplete();
}

void TcpConnection::OnWriteComplete(uv_write_t *req, int32_t status)
{
    WriteReq *write_req = (WriteReq *)req;
    TcpConnection *ptr = (TcpConnection *)req->data;
    ptr->HandleWriteComplete(write_req, status);
}

void TcpConnection::OnDisConnect()
{
    printf("TcpConnection::OnDisConnect this:%p\n", this);
    SetConState(kDisconnected);
    if(con_close_cb_) {
        con_close_cb_(shared_from_this());
    }
}

void TcpConnection::DoCloseComplete()
{
    SetConState(kDisconnected);
    if(close_complete_cb_) {
        close_complete_cb_();
    }
}

int32_t TcpConnection::Write(const WriteReqSptr& req)
{
    if(state_ != kConnected) {
        if(req->cb) {
            WriteInfo info{};
            info.status = -1;
            info.buf = req->uv_buf.base;
            info.size = req->uv_buf.len;
            info.user_data = req->user_data;
            req->cb(info);
        }
        printf("TcpConnection::Send error state_:%d\n", (int32_t)state_);
        return -1;
    }

    int32_t ret = uv_write((uv_write_t *)req.get(), (uv_stream_t *)tcp_handle_.get(), &req->uv_buf, 1, OnWriteComplete);
    if(ret != 0) {
        printf("TcpConnection::Send ret=%d error: %s\n", ret, uv_strerror(ret));
        return -1;
    }

    write_reqs_.push_back(req);

    return 0;
}

int32_t TcpConnection::HandleRead(ssize_t nread)
{
    if(nread > 0 ) {
        read_buffer_->HasWritten(nread);
        printf("TcpConnection::OnRead %.*s\n", (int32_t)read_buffer_->ReadableBytes(), read_buffer_->Peek());
        if(message_cb_) {
            message_cb_(shared_from_this(), read_buffer_.get());
        }
    } else if(nread < 0) {
        printf("TcpConnection::OnRead nread=%ld error: %s\n", nread, uv_strerror(nread));
        OnDisConnect();
    }

    return 0;
}

int32_t TcpConnection::HandleWriteComplete(WriteReq *req, int32_t status)
{

    printf("HandleWriteComplete status=%d len:%zu\n", status, req->uv_buf.len);

    if(write_reqs_.empty()) {
        printf("HandleWriteComplete error write_reqs_.empty()\n");
        return 0;
    }
    WriteReqSptr write_req = write_reqs_.front();
    write_reqs_.pop_front();

    if((intptr_t)req != (intptr_t)write_req.get()){
        printf("HandleWriteComplete error req:%p write_req.get():%p\n", req, write_req.get());
    }

    if(write_req->cb) {
        WriteInfo info{};
        info.status = status;
        info.buf = req->uv_buf.base;
        info.size = req->uv_buf.len;
        info.user_data = write_req->user_data;
        write_req->cb(info);
    }

    return 0;
}

} //namespace 
} //namespace 